Streamlining Access to European Electricity Market Data
University of Duisburg-Essen, House of Energy, Climate, and Finance
December 9, 2025
Methodological
Applied:
ENTSOE announced to shutdown SFTP server end of September 2025
Later postponed until “mid November”
We queried that server hourly to obtain:
Remaining options
Obtain CSV Extracts via RESTful API
Obtain data as XML via RESTful API
In a nutshell:
Technical Overview
Used to build a query.
We mirror ENTSOE’s documentation:
entsoe.Market.ActualTotalLoad
entsoe.Market.DayAheadTotalLoadForecast
entsoe.Market.WeekAheadTotalLoadForecast
entsoe.Market.MonthAheadTotalLoadForecast
entsoe.Market.YearAheadTotalLoadForecast
entsoe.Market.YearAheadForecastMargin
entsoe.Load.<...>
entsoe.Generation.<...>
entsoe.Transmission.<...>
entsoe.Outages.<...>
entsoe.Balancing.<...>
entsoe.Master Data.<...>
entsoe.OMI.<...>Each class holds .params dictionary
Each class provides .query method.
entsoe.utils.extract_records
entsoe.utils.add_timestamps
entsoe.utils.calculate_timestamp
entsoe.utils.mappings = {
"10YBE----------2": {"BE": ["BZN", "CTA", "CTY", "LFA", "LFB", "MBA", "SCA"]},
"10YIE-1001A00010": {"IE": ["CTA", "CTY", "SCA"], "SEM(EirGrid)": ["MBA"]},
"10Y1001A1001A59C": {
"IE(SEM)": ["BZN", "MBA", "SCA"],
"IE-NIE": ["LFB"],
"Ireland": ["SNA"],
},
}
entsoe.codes.<...> # Enums for translationInternal modules: entsoe.xml_models, entsoe.query
The package will read API key from ENTSOE_API environment variable
from pandas import DataFrame
from entsoe.Market import EnergyPrices
from entsoe.utils import extract_records, add_timestamps
# from entsoe.config import set_config
# set_config(security_token=None, endpoint_url=None, timeout=5, retries=5,
# retry_delay=lambda attempt: 2**attempt, max_workers=4, log_level="SUCCESS"
# )
result = EnergyPrices(
in_domain="10Y1001A1001A82H", # Germany, DE-LU Biddingzone
out_domain="10Y1001A1001A82H",
period_start=202012312300,
period_end=202101022300,
).query_api()
records = extract_records(result) # Serialize <=> Convert from pydantic class to dict
records = add_timestamps(records)
df = DataFrame(records)httpxloguru
entsoe.config.set_config(loglevel = "DEBUG")offset parameter).zip responses@split_date_range
@pagination
def query_api(params):
results = query_and_parse(params)
return results
@retry
def query_and_parse(params):
responses = fetch_responses(params)
xml_models = parse_response(responses)
return xml_models
@unzip
def fetch_responses(params) -> list[Response]:
response = query_core(params)
return [response]
@handle_acknowledgement
def parse_response(response):
_, matching_class = extract_namespace_and_find_classes(response)
xml_model = XmlParser().from_string(response.text, matching_class)
return xml_model@check_if_banned
@check_service_unavailable
@rate_limit(max_calls=380, period=60)
def query_core(params):
params_with_token = {**params, "securityToken": get_config().security_token}
response = get(config.endpoint_url, params=params_with_token, timeout=config.timeout)
return responsedef check_service_unavailable(func):
@wraps(func)
def service_unavailable_wrapper(*args, **kwargs):
response = func(*args, **kwargs)
if response.status_code == 503:
# @retry decorator will check for this error
raise ServiceUnavailableError("ENTSO-E API is unavailable (HTTP 503).")
return response
return service_unavailable_wrapperDecorators create a hidden layer to handle exceptions, without cluttering the core logic.
Roadmap: